package org.zjt.chapter1.http;

import com.alibaba.fastjson.JSON;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.collection.LongObjectHashMap;
import io.netty.util.concurrent.*;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zjt.chapter1.http.domain.CallBackTextWebSocketFrame;
import org.zjt.chapter1.http.domain.RequestMsg;

import java.net.URI;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Description:
 *
 *
 * WebSocket类型：
         BinaryWebSocketFrame，      包含二进制数据
         TextWebSocketFrame，        包含文本数据
         ContinuationWebSocketFrame，包含二进制数据或文本数据，BinaryWebSocketFrame和TextWebSocketFrame的结合体
         CloseWebSocketFrame，       WebSocketFrame代表一个关闭请求，包含关闭状态码和短语
         PingWebSocketFrame，        WebSocketFrame要求PongWebSocketFrame发送数据
         PongWebSocketFrame，        WebSocketFrame要求PingWebSocketFrame响应
 *
 *
 *
 * @author juntao.zhang
 * Date: 2018-07-19 下午5:15
 * @see
 */
public class WebSocketServer extends AbstractHttpServer {

    public static void main(String[] args) {
        new WebSocketServer().start(8943);
    }


    @Override
    public ChannelInitializer<NioSocketChannel> getChannelInitializer() {
        return new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();

                pipeline.addLast(new HttpServerCodec());
                pipeline.addLast(new HttpObjectAggregator(12 << 10));

                pipeline.addLast(new HttpContentCompressor());

                pipeline.addLast(new HttpServerExpectContinueHandler());

                // upgrade 升级为 ws
                pipeline.addLast(new WebSocketServerProtocolHandler("/websocket"));


                // ws 文本处理器
                pipeline.addLast(new TextFrameHandler());


                //pipeline.addLast(new ConsumerAgentHttpServerFastHandler());

                pipeline.addLast(new SimpleChannelInboundHandler<HttpRequest>() {
                    @Override
                    protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {

                        ByteBuf byteBuf = Unpooled.wrappedBuffer("hello world".getBytes());
                        DefaultFullHttpResponse httpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,HttpResponseStatus.OK,byteBuf);

                        HttpHeaders headers = httpResponse.headers();
                        headers.add("Content-Type","text/plain");

                        // http body 字节数
                        headers.addInt("content-length",byteBuf.readableBytes());

                        // 告知请求方可以建立长连接
                        headers.add("Connection","keep-alive");

                        boolean keepAlive = HttpUtil.isKeepAlive(msg);
                        if (keepAlive){
                            ctx.writeAndFlush(httpResponse);
                        }else {
                            ctx.writeAndFlush(httpResponse).addListener(ChannelFutureListener.CLOSE);
                        }
                    }
                });
            }
        };
    }


    /**
     * ws 文本处理
     */
    @Slf4j
    public static class TextFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame>{

        public static FastThreadLocal<Promise> promiseHolder = new FastThreadLocal<>() ;



        private static final String apiKey = System.getProperty("apiKey");
        private static final String userId = System.getProperty("userId");

        private static final String URL = "http://localhost:8080";

        private Channel channel;

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            reConnect(ctx);
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            log.info("-------------userEventTriggered--------------------:{}",evt);
            if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete ){
                ctx.writeAndFlush(new TextWebSocketFrame("连接成功!"));
            }else{
                super.userEventTriggered(ctx, evt);
            }
        }

        private void reConnect(ChannelHandlerContext ctx) {
             channel = new Bootstrap().channel(NioSocketChannel.class)
                    .group(ctx.channel().eventLoop())
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {

                            // 客户端接收到的是httpResponse响应，所以要使用HttpResponseDecoder进行解码
                            ch.pipeline().addLast(new HttpResponseDecoder());

                            // 客户端发送的是httprequest，所以要使用HttpRequestEncoder进行编码
                            ch.pipeline().addLast(new HttpRequestEncoder());
                            ch.pipeline().addLast(new HttpObjectAggregator(12 << 10));

                            ch.pipeline().addLast(new ConsumerAgentClientHandler());
                        }
                    })
                    .connect("localhost", 8080).channel();
        }



        @Override
        protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {


            String text = msg.text();
            log.info("text :{}",text);
            RequestMsg.InputText inputText = RequestMsg.InputText.builder().text(text).build();
            RequestMsg.Perception perception = RequestMsg.Perception.builder().inputText(inputText).build();
            RequestMsg.UserInfo userInfo = RequestMsg.UserInfo.builder().apiKey(apiKey).userId(userId).build();
            RequestMsg requestMsg = RequestMsg.builder().perception(perception).reqType(1).userInfo(userInfo).build();

            String conn = JSON.toJSONString(requestMsg);


            DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, new URI(URL).toASCIIString(), Unpooled.wrappedBuffer(conn.getBytes("UTF-8")));



            HttpHeaders headers = request.headers();

            headers.add("Content-Type","text/plain");

            // http body 字节数
            headers.addInt("content-length",request.content().readableBytes());

            // 告知请求方可以建立长连接
            headers.add("Connection","keep-alive");


            channel.writeAndFlush(request);


            // 写出一个回调函数，
            Promise<String> agentResponsePromise = new DefaultPromise<String>(ctx.executor());
            agentResponsePromise.addListener(future -> {
                String s = (String) future.get();
                ctx.channel().writeAndFlush(new TextWebSocketFrame(s+""));
            });

            //保存回调
            promiseHolder.set(agentResponsePromise);

        }
    }

    /**
     * ws  二进制处理
     */
    @Slf4j
    public static class BinaryFrameHandler extends SimpleChannelInboundHandler<BinaryWebSocketFrame>{

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, BinaryWebSocketFrame msg) throws Exception {

        }
    }

    /**
     * ws 文本和二进制结合体
     */
    @Slf4j
    public static class ContinuationFrameHandler extends SimpleChannelInboundHandler<ContinuationWebSocketFrame> {

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ContinuationWebSocketFrame msg) throws Exception {

        }
    }


    public static class ConsumerAgentClientHandler extends SimpleChannelInboundHandler<Object> {

        public ConsumerAgentClientHandler() {
        }

        private Logger logger = LoggerFactory.getLogger(ConsumerAgentClientHandler.class);

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
            callback(msg);
        }

        private void callback(Object agentResponse) {
            if (agentResponse instanceof FullHttpResponse){
                FullHttpResponse httpResponse = FullHttpResponse.class.cast(agentResponse);
                ByteBuf content = httpResponse.content();

                byte[] bytes = new byte[content.readableBytes()];
                content.readBytes(bytes);

                Promise promise = TextFrameHandler.promiseHolder.get();
                if(promise !=null){
                    promise.trySuccess(new String(bytes));
                }
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            logger.error("consumerAgentHandler出现异常", cause);
        }

    }

}
